package edu.hit.crawler;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.DefaultHttpClient;

import edu.hit.crawler.io.CrawItem;
import edu.hit.crawler.util.URLCanonicalizer;
import edu.hit.crawler.util.URLFilter;

@SuppressWarnings("deprecation")
public class UpdateDb extends Configured implements Tool {
	private static long start = System.currentTimeMillis();
	private static int count = 0;

	public static class UpdateDbMapper extends
			Mapper<Text, CrawItem, Text, CrawItem> {
		@Override
		public void map(Text key, CrawItem value, Context context)
				throws IOException, InterruptedException {

			String url = value.getUrl();
			if (0 == url.length())
				return;
			if (!URLFilter.filter(url))
				return;
			url = URLCanonicalizer.getCanonicalURL(url);

			value.setUrl(url);
			if (url != null && url.length() > 0) {
				context.write(new Text(url), value);
			}
		}
	}

	public static class UpdateDbReducer extends
			Reducer<Text, CrawItem, Text, CrawItem> {
		String uiservlet;
		int reportInterval;

		@Override
		protected void setup(
				org.apache.hadoop.mapreduce.Reducer<Text, CrawItem, Text, CrawItem>.Context context)
				throws IOException, InterruptedException {
			uiservlet = context.getConfiguration().get(
					"org.work.crawler.ui.servlet");
			reportInterval = context.getConfiguration().getInt("org.work.crawler.reportInterval", 2*1000);


		};

		@Override
		public void reduce(Text key, Iterable<CrawItem> values, Context context) {
			count++;
			long now = System.currentTimeMillis();
			if (now - start > reportInterval) {
				try {
					DefaultHttpClient httpclient = new DefaultHttpClient();
					String uri = uiservlet + "?state=update&count="
							+ Integer.toString(count);
					HttpPost post = new HttpPost(uri);
					HttpResponse response = null;
					response = httpclient.execute(post);

					System.out.println(response.getStatusLine());
					httpclient.getConnectionManager().shutdown();

					count = 0;
					start = now;
				} catch (ClientProtocolException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			boolean isCrawed = false;
			boolean isGone = false;
			boolean isNotFound = false;
			boolean isRetry = false;

			int statu = CrawItem.UNCRAWL;
			int outlinks = 0;
			for (CrawItem value : values) {
				statu = value.getStatus();

				if (statu == CrawItem.SUCCESS) {
					outlinks = value.getOutlinks();
					isCrawed = true;
					break;
				} else if (statu == CrawItem.RETRY) {
					isRetry = true;
				} else if (statu == CrawItem.GONE) {
					isGone = true;
				} else if (statu == CrawItem.NOTFOUND) {
					isNotFound = true;
				}
			}

			try {

				if (isCrawed) {
					context.write(key, new CrawItem(key.toString(),
							CrawItem.SUCCESS, outlinks));
				} else if (isRetry) {
					context.write(key, new CrawItem(key.toString(),
							CrawItem.RETRY));
				} else if (isGone) {
					context.write(key, new CrawItem(key.toString(),
							CrawItem.GONE, 0));
				} else if (isNotFound) {
					context.write(key, new CrawItem(key.toString(),
							CrawItem.NOTFOUND, 0));
				} else {
					context.write(key, new CrawItem(key.toString(), statu));
				}

			} catch (IOException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		int res = ToolRunner.run(conf, new UpdateDb(), args);
		System.exit(res);
	}

	@Override
	public int run(String[] args) throws Exception {

		Configuration conf = getConf();
		Job job = new Job(conf, "updateDb");
		job.setJarByClass(UpdateDb.class);

		job.setMapperClass(UpdateDbMapper.class);
		job.setReducerClass(UpdateDbReducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(CrawItem.class);

		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(MapFileOutputFormat.class);

		job.setNumReduceTasks(6);

		String workdir = conf.get("org.work.crawler.dir", "crawler//");

		/*
		 * // unselcet FileInputFormat .addInputPath(job, new Path(workdir +
		 * "unselected_tmp/"));
		 */

		// the new outlinks
		long latest = 0;
		Path inPath = null;
		FileSystem fs = FileSystem.get(conf);
		for (FileStatus stat : fs.listStatus(new Path(workdir + "tmp/"))) {
			if (stat.getModificationTime() > latest) {
				inPath = stat.getPath();
				latest = stat.getModificationTime();
			}
		}
		if (null != inPath)
			FileInputFormat.addInputPath(job, inPath);

		/*
		 * inPath = null; latest = 0; // the original crawl_db for (FileStatus
		 * stat : fs.listStatus(new Path(workdir + "crawl_db/"))) { if
		 * (stat.getModificationTime() > latest) { inPath = stat.getPath();
		 * latest = stat.getModificationTime(); } }
		 */
		FileInputFormat.addInputPath(job, new Path(workdir + "crawl_db/"));

		Path out = new Path(workdir + "crawl_db_tmp/");

		SequenceFileOutputFormat.setOutputPath(job, out);

		int res = job.waitForCompletion(true) ? 0 : 1;

		if (res == 0) {

			fs = FileSystem.get(conf);

			if (fs.exists(new Path(workdir + "crawl_db"))) {
				fs.delete(new Path(workdir + "crawl_db/"), true);
				fs.rename(out, new Path(workdir + "crawl_db/"));
			}
			System.out.println("Update done!");
		} else {
			System.out.println("Update failed.");
		}

		return res;
	}

}
